package cn.lianxi;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;

/**
 * 4、有一个hdfs数据目录/tmp/data/中，存放若干txt文本文件，文件内部存放的是由空格分隔的中英文单词，求该目录下所有单词的出现频率，
 * 
 * 并按出现频率倒序排列后输出到控制台中。  通过java api读取hdfs的文件目录/tmp/data下的所有hdfs路径，存入一个set当中。 
 * 
 * 通过java api代码读取set中每个元素即hdfs路径的文本内容。 
 * 
 * 将每个文本内容按空格分隔开，放入map<String,Integer>结构中。如不存则value置1，如果存在，则value+1操作。 
 * 
 * 将map转换成list后，写list比较器，按其元素的value进行排序。 
 * 
 * 将排序后结果通过java标准输出打印到控制台当中。
 * 
 * 
 * @author yingfing
 *
 */
public class HDFSLianXi_04 {

	public static void main(String[] args) throws Exception {

		TreeSet<String> fisSet = new TreeSet<String>();
		Map<String, Integer> pathMap = new TreeMap<String, Integer>();

		// 0. 获取集群配置信息
		Configuration conf = new Configuration();
		// 1. 获取文件系统
		FileSystem fs = FileSystem.get(conf);

		// 2. 获取所有的 .txt 文件列表
		Path path1 = new Path("/user/zhaoyingfei/tmp/data");
		FileStatus[] fss = fs.listStatus(path1);

		// 遍历放入 set
		for (FileStatus fis : fss) {
			fisSet.add(fis.getPath().toString().substring(27));
		}
		for (String srt : fisSet) {
			// System.out.println(srt);
			Path path2 = new Path(srt);
			FSDataInputStream in = fs.open(path2);
			// 3. 读取文件内容
			InputStreamReader isr = new InputStreamReader(in);
			BufferedReader br = new BufferedReader(isr);
			String str = null;
			while ((str = br.readLine()) != null) {
				String[] strarray = str.split(" ");
				for (String string : strarray) {
					if (pathMap.containsKey(string)) {
						pathMap.put(string, pathMap.get(string) + 1);
					} else {
						pathMap.put(string, 1);
					}
				}
			}
		}

		// 先把entry拿出来
		Set set = pathMap.entrySet();

		// 调用ArrayList 的有参构造,把 set 直接转换为 List
		List<Entry<String, Integer>> list = new ArrayList<Entry<String, Integer>>(
				set);

		Collections.sort(list, new Comparator<Entry<String, Integer>>() {

			@Override
			public int compare(Entry<String, Integer> o1,
					Entry<String, Integer> o2) {
				return o2.getValue() - o1.getValue();
			}
		});
		System.out.println(list);
		// 5.关闭流
		fs.close();

	}

}
